package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo04DSL {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo04DSL")
      .master("local")
      // 默认200 避免shuffle过后产生过多的分区 导致生成的task数量过多
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()

    // 导入function以及隐式转换
    import org.apache.spark.sql.functions._
    import spark.implicits._

    // DSL的常用方法
    /**
     * SQL中常用的一些关键字：
     * select 查询
     * from tb
     * where 过滤
     * group by 分组 --> max、min、avg、sum、count 聚合函数
     * having 分组聚合后的过滤
     * order by 排序
     * limit 限制返回的数据条数
     * 关联：
     * 内连接：inner join、join
     * 外连接：left join、right join
     * 全外连接：full outer join （很少使用）
     * 连接：
     * union、union all
     * 去重：distinct
     * 常用的函数：
     * 字符串函数、数值函数、日期函数、窗口函数、条件函数
     * */

    // 构建DataFrame
    val stuDF: DataFrame = spark
      .read
      .format("csv") // 文本文件统一用csv去读取
      .option("sep", ",") // 最好在读取文本类文件的时候加上sep分隔符
      .schema("id String,name String,age Int,gender String,clazz String") // 给数据附上结构（列名及列的类型）
      .load("Spark/data/students.txt") // 指定读取文件的路径

    // 对多次使用的DataFrame进行cache
    stuDF.cache()

    // 打印DataFrame(表)结构
    stuDF.printSchema()

    // select
    stuDF.select("id", "name", "age") // 字符串表达式
    //      .show()

    stuDF.select($"id", $"name", $"age") // 使用 $ 取列 ==> 列表达式 (推荐)
    //      .show()

    // where、filter过滤
    // 过滤 文科班 的学生
    stuDF.where("clazz like '文科%'") // 字符串表达式
    //      .show()

    stuDF.where($"clazz" like "文科%") // 列表达式
    //      .show()

    // 使用filter 既支持 字符串表达式 也支持 列表达式 还支持 函数式编程的方式
    // DataFrame 跟 DataSet[Row] 没有区别
    // DataSet 相比 RDD 会有更多好用的API可以快速完成数据分析
    // DataFrame中的数据实际上都是一个个的Row的对象 是有结构
    stuDF.filter(row => row.getAs[String]("clazz").contains("文科"))
    //      .show()


    // 用完记得释放缓存
    stuDF.unpersist()

    // 统计班级人数
    /**
     * groupBy会产生shuffle
     * 在Spark SQL中默认shuffle过后的分区数默认为 200 那么会产生200个任务
     * 可以通过 spark.sql.shuffle.partitions 进行配置
     */
    stuDF
      .groupBy($"clazz")
      .agg(count($"id") as "cnt")
    //      .show()

    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id") as "cnt") // 注意数据有重复 需要使用countDistinct
    //      .show()

    // having 分组聚合后的过滤
    // 统计班级人数 并过滤出 人数大于90的班级
    /**
     * 为什么在DSL中聚合后可以使用where过滤，在SQL中只能用having？
     * 在Spark SQL的DSL中会按照从头往后的顺序执行
     *
     */
    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id") as "cnt")
      .where($"cnt" > 90)
    //      .show()

    // orderBy limit
    // 统计班级人数 并过滤出 人数大于90的班级 并按照人数从大到小进行排列
    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id") as "cnt")
      .where($"cnt" > 90)
      // 在Spark SQL中 orderBy跟sort是一样的 理论上都是做全局排序
      // 如果不是可以通过改变分区数做到全局排序
      //      .coalesce(1)
      .orderBy($"cnt".desc)
      .limit(3)
    //      .show()

    // 联接 相当于union
    val stuSampleDF1: Dataset[Row] = stuDF.sample(0.01, seed = 1)
    val stuSampleDF2: Dataset[Row] = stuDF.sample(0.01, seed = 1)

    //    stuSampleDF1.show()
    //    stuSampleDF2.show()
    // 只有格式一致的两份DataFrame才可以进行联接
    /**
     * SQL中的union会进行去重、union all不会
     * Spark SQL中的union实际上是RDD中的union方法 不会进行去重
     * 如果需要去重 可以接distinct
     */
    stuSampleDF1
      .union(stuSampleDF2)
    //      .show()

    // 通过union+distinct实现SQL中union可以去重的效果
    stuSampleDF1
      .union(stuSampleDF2)
      .distinct()
    //      .show()


    // 关联
    val scoDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,sub_id String,score Int")
      .load("Spark/data/score.txt")

    // inner join
    stuDF
      // 两个DF在关联时如果关联字段相同 使用列表达式会有问题
      .join(scoDF, "id")
    //      .show()

    // left join、right join
    // joinType可以指定：
    // inner, cross, outer, full, full_outer, left, left_outer, right, right_outer, left_semi, left_anti
    // 一定要注意 关联的字段名相同时 又需要指定关联的方式 必须将关联的字段放入List中 作为参数传入
    stuDF
      .join(scoDF, List("id"), "left")
    //      .show()

    // 将scoDF中的id列改名为sid
    val newScoDF: DataFrame = scoDF.withColumnRenamed("id", "sid")
    newScoDF
      .join(stuDF, $"id" === $"sid", "right")
    //      .show()

    // DSL中实现case when
    // 计算每个学生的平均分 考虑各科目总分不一样的情况（数值归一化）
    // 如果平均分低于60    则 输出不及格
    // 如果平均分大于等于60 则 输出及格
    // 如果平均分大于等于70 则 输出中等
    // 如果平均分大于等于80 则 输出良好
    // 如果平均分大于等于90 则 输出优秀

    // 读取科目表并构建DataFrame
    val subDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("sub_id String,sub_name String,sub_score Int")
      .load("Spark/data/subject.txt")

    // 基于科目表 将所有学生的分数转换为百分制
    scoDF
      .join(subDF, "sub_id")
      .select($"id", $"sub_id", round($"score" * 100.toDouble / $"sub_score", 2) as "new_score")
      .groupBy($"id")
      // 计算平均分
      .agg(round(avg($"new_score"), 2) as "avg_score")
      // 实现case when
      .select($"id", $"avg_score",
        when($"avg_score" >= 90, "优秀")
          .when($"avg_score" >= 80, "良好")
          .when($"avg_score" >= 70, "中等")
          .when($"avg_score" >= 60, "及格")
          .otherwise("不及格") as "等级"
      )
      .sort($"avg_score".desc)
      .show()

    /**
     * 基于dianxin数据统计：
     * 每个城市游客数量最多的前三个区县
     *
     */

    val dianXinDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("mdn String,grid_id Bigint,city_id Int,county_id Int,duration Int,start_time String,end_time String,pt String")
      .load("Spark/data/dianxin.csv")
    dianXinDF.printSchema()
    dianXinDF.show()

    // SQL的方式
    dianXinDF.createOrReplaceTempView("dianxin")

    spark.sql(
      """
        |select  tt1.city_id
        |        ,tt1.county_id
        |        ,tt1.cnt
        |        ,tt1.rn
        |from (
        |    select  t1.city_id
        |            ,t1.county_id
        |            ,t1.cnt
        |            ,row_number() over(partition by t1.city_id order by t1.cnt desc) as rn
        |    from (
        |        SELECT  city_id
        |                ,county_id
        |                ,count(distinct mdn) as cnt
        |        from dianxin
        |        group by city_id,county_id
        |    ) t1
        |) tt1 where tt1.rn<=3
        |""".stripMargin).show(50)

    // DSL的方式
    dianXinDF
      .groupBy($"city_id", $"county_id")
      .agg(countDistinct($"mdn") as "cnt")
      .select($"city_id", $"county_id", $"cnt"
        , row_number() over Window.partitionBy($"city_id").orderBy($"cnt".desc) as "rn")
      .where($"rn" <= 3)
      .show(50)

    dianXinDF
      .groupBy($"city_id", $"county_id")
      .agg(countDistinct($"mdn") as "cnt")
      .withColumn("rn",row_number() over Window.partitionBy($"city_id").orderBy($"cnt".desc))
      .where($"rn" <= 3)
      .show(50)


  }

}
